package com.cti.commons.delayqueue.core;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RHyperLogLog;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author et.wang
 * @description: 存放可以消费的任务
 * @date 2022/10/27 19:36
 */
@Component
public class ReadyQueue {

    private RedissonClient redissonClient;

    private RHyperLogLog<String> processedMessageIds;

    @Autowired
    public void setRedissonClient(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
        this.processedMessageIds = redissonClient.getHyperLogLog("delayQueue:processedMessageIds");
    }

    /**
     * 添加任务Id到准备队列
     *
     * @param topic
     * @param messageId
     */
    public void pushToReadyQueue(String topic, Long messageId) {
        RBlockingQueue<Long> blockingQueue = redissonClient.getBlockingQueue(topic);
        String messageIdString = messageId.toString();
        if (!processedMessageIds.add(messageIdString)) {
            // messageId已经存在，表示已处理过，不进行添加到队列操作
            return;
        }
        blockingQueue.offer(messageId);
    }

    /**
     * 从准备队列中获取任务Id
     *
     * @param topic
     * @return messageId
     */
    public Long pollFormReadyQueue(String topic) {
        RBlockingQueue<Long> blockingQueue = redissonClient.getBlockingQueue(topic);
        return blockingQueue.poll();
    }

}
